package com.example.yian.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.aliyuncs.utils.StringUtils;
import jakarta.websocket.*;
import jakarta.websocket.server.PathParam;
import jakarta.websocket.server.ServerEndpoint;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.InputStreamResource;
import org.springframework.http.HttpInputMessage;
import org.springframework.http.HttpOutputMessage;
import org.springframework.http.MediaType;
import org.springframework.http.converter.AbstractHttpMessageConverter;
import org.springframework.http.converter.HttpMessageNotReadableException;
import org.springframework.http.converter.HttpMessageNotWritableException;
import org.springframework.stereotype.Component;
import org.springframework.util.StreamUtils;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.charset.Charset;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

public class CustomInputStreamResourceHttpMessageConverter extends AbstractHttpMessageConverter<InputStreamResource> {

    public CustomInputStreamResourceHttpMessageConverter() {
        super(new MediaType("application", "octet-stream", Charset.forName("UTF-8")));
    }

    @Override
    protected boolean supports(Class<?> clazz) {
        return InputStreamResource.class.isAssignableFrom(clazz);
    }

    @Override
    protected InputStreamResource readInternal(Class<? extends InputStreamResource> clazz, HttpInputMessage inputMessage) throws IOException, HttpMessageNotReadableException {
        InputStream inputStream = inputMessage.getBody();

        // 读取 InputStream 中的所有数据到 byte 数组
        byte[] bytes = StreamUtils.copyToByteArray(inputStream);

        // 使用 byte 数组创建 ByteArrayInputStream
        InputStream byteStream = new ByteArrayInputStream(bytes);

        // 创建 InputStreamResource 并返回
        return new InputStreamResource(byteStream);
    }

    @Override
    protected void writeInternal(InputStreamResource inputStreamResource, HttpOutputMessage outputMessage) throws IOException, HttpMessageNotWritableException {
        InputStream inputStream = inputStreamResource.getInputStream();
        OutputStream outputStream = outputMessage.getBody();
        StreamUtils.copy(inputStream, outputStream);
    }

    @Slf4j
    @ServerEndpoint("/websocket/{userId}")
    @Component
    public static class WSServer {

        private static final Logger log = (Logger) LoggerFactory.getLogger(WSServer.class);

        /**
         * 当前在线连接数
         */
        private static AtomicInteger onlineCount = new AtomicInteger(0);

        /**
         * 用来存放每个客户端对应的 WebSocketServer 对象
         */
        private static ConcurrentHashMap<String, WSServer> webSocketMap = new ConcurrentHashMap<>();

        /**
         * 与某个客户端的连接会话，需要通过它来给客户端发送数据
         */
        private Session session;

        /**
         * 接收 userId
         */
        private String userId = "";

        /**
         * 连接建立成功调用的方法
         */
        @OnOpen
        public void onOpen(Session session, @PathParam("userId") String userId) {
            this.session = session;
            this.userId = userId;
            if (webSocketMap.containsKey(userId)) {
                webSocketMap.remove(userId);
                webSocketMap.put(userId, this);
            } else {
                webSocketMap.put(userId, this);
                addOnlineCount();
            }
            log.info("用户连接:" + userId + ",当前在线人数为:" + getOnlineCount());
            try {
                sendMessage("连接成功！");
            } catch (IOException e) {
              // log.log("用户:" + userId + ",网络异常!!!!!!");
            }
        }

        /**
         * 连接关闭调用的方法
         */
        @OnClose
        public void onClose() {
            if (webSocketMap.containsKey(userId)) {
                webSocketMap.remove(userId);
                subOnlineCount();
            }
            log.info("用户退出:" + userId + ",当前在线人数为:" + getOnlineCount());
        }

        /**
         * 收到客户端消息后调用的方法
         *
         * @param message 客户端发送过来的消息
         */
        @OnMessage
        public void onMessage(String message, Session session) {
            log.info("用户消息:" + userId + ",报文:" + message);
            if (!StringUtils.isEmpty(message)) {
                try {
                    JSONObject jsonObject = JSON.parseObject(message);
                    jsonObject.put("fromUserId", this.userId);
                    String toUserId = jsonObject.getString("toUserId");
                    if (!StringUtils.isEmpty(toUserId) && webSocketMap.containsKey(toUserId)) {
                        webSocketMap.get(toUserId).sendMessage(jsonObject.toString());
                    } else {
                       // log.log("请求的 userId:" + toUserId + "不在该服务器上");
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }

        /**
         * 发生错误时调用
         *
         * @param session
         * @param error
         */
        @OnError
        public void onError(Session session, Throwable error) {
            //log.log("用户错误:" + this.userId + ",原因:" + error.getMessage());
            error.printStackTrace();
        }

        /**
         * 实现服务器主动推送
         */
        public void sendMessage(String message) throws IOException {
            this.session.getBasicRemote().sendText(message);
        }

        public static synchronized AtomicInteger getOnlineCount() {
            return onlineCount;
        }

        public static synchronized void addOnlineCount() {
            WSServer.onlineCount.getAndIncrement();
        }

        public static synchronized void subOnlineCount() {
            WSServer.onlineCount.getAndDecrement();
        }
    }
}